/**
 * Copyright (c) 2020 kedacom
 * OpenATC is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 * http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 **/

package com.openatc.agent.handler;

import com.google.gson.*;
import com.openatc.agent.controller.TemplateController;
import com.openatc.agent.gateway.WebSocketServer;
import com.openatc.agent.model.CountDownPattern;
import com.openatc.agent.model.CountDownStage;
import com.openatc.agent.model.Fault;
import com.openatc.agent.service.*;
import com.openatc.comm.data.MessageData;
import com.openatc.comm.handler.ICommHandler;
import com.openatc.core.model.RESTRet;
import com.openatc.core.util.DateUtil;
import com.openatc.core.util.GsonUtil;
import com.openatc.model.model.*;
import org.apache.kafka.common.KafkaException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import javax.websocket.Session;
import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;

import static com.openatc.comm.common.CosntDataDefine.*;
import static com.openatc.model.model.SystemEvent.EVENT_LEVEL_URGENT;
import static com.openatc.model.model.SystemEvent.TYPE_SYSTEM;
import static com.openatc.model.model.SystemEventDetails.*;

/**
 * @ClassName AgentHandler
 * @Auther zhangwenchao
 * @CreateTime 2019/11/5 16:09
 * @Description
 */
@Component
public class AgentHandler extends ICommHandler {

    private static Logger logger = Logger.getLogger(AgentHandler.class.toString());

    @Value("${spring.redis.enable}")
    private boolean isRedisEnable;
    @Autowired
    @Qualifier("pdKafkaTemplate")
    private KafkaTemplate<String, String> kafkaTemplate;
    @Value("${kafka.topic.push.statuspattern}")
    private String statusPatternPushTopic;
    @Value("${kafka.topic.push.laneflow}")
    private String laneFlowPushTopic;
    @Value("${statuspattern.pushkafka.enable}")
    private boolean isPushKafkaEnable;
    @Autowired
    private JdbcTemplate jdbcTemplate;
    @Autowired
    private FaultDao faultDao;
    @Autowired
    public StringRedisTemplate stringRedisTemplate;
    @Autowired
    private FaultService faultService;
    @Autowired
    private AscsDao ascsDao;
    @Autowired
    private HistoryDataDao historyDataDao;
    @Autowired
    private DevService devService;
    @Autowired
    private TemplateController templateController;
    @Autowired
    private FeatureService featureService;

    public static BlockingQueue<MessageData> MdQueue = new LinkedBlockingQueue<>();
    private static final Map<String,Integer> lightStatusFaultAgent = new HashMap<>();

    public static int MdTodayCount = 0; // 今日收到消息数
    public static int MdminQps = 0; // 5分钟QPS
    public static Queue<String> patternsqlqueue = new ConcurrentLinkedQueue<>(); // 批量保存的方案sql缓存

    private String agenttype = "asc";

    private Gson gson = new Gson();
    Map<Session, WebSocketServer> patternWebSocketSet = WebSocketServer.getPatternWebSocketSet();
    @Autowired
    private WebSocketServer webSocketServer;

    @Override
    public void process(MessageData msg) {

        try {
            MdQueue.put(msg);
            MdTodayCount++;
            MdminQps++;
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    @Async("taskExecutor")
    public void processExec() {
        while (true) {
            try {
                MessageData msg = null;
                try {
                    msg = MdQueue.take();
                } catch (InterruptedException e) {
                    continue;
                }
                // 如果消息为空，则返回
                if (msg == null) {
                    logger.warning("AgentHandler process: MessageData is null");
                    continue;
                }

                // 检查消息时间
                Date curDate = new Date();
                String curtime = DateUtil.date2esstr(curDate);
                msg.setSystime(curtime);
                msg.setSts(curDate.getTime());

                if (msg.getCreatetime() == null) {
                    msg.setCreatetime(curtime);
                }

                // 获取消息类型
                String infotype = msg.getInfotype();
                if (infotype == null) {
                    logger.warning("AgentHandler process: infoType is null");
                    continue;
                }

                // 获取操作类型
//                String operation = msg.getOperation();

                String agentid;

                // 设备的注册消息
                if (login.equals(infotype)) {
//                    logger.info("Recv LoginMsg :" + msg );

                    agentid = ascsDao.processLoginMsg(msg);
                    if (agentid == null){
                        logger.warning("LoginMsg cannot find agentid! msg:" + msg);
                        continue;
                    }
                    // 把消息放入定时器的消息队列
                    ascsDao.putMsgToLoginMap(agentid,msg);
                    // redis发布
                    String loginKey = agenttype + ":" + infotype + ":" + agentid;
                    saveAndPublishMsg(msg,loginKey,40);

                    // 保持全息设备在线
                    pushMessageToKafka("openatcstatusheartbeat",msg);

                } else {
                    if (msg.getAgentid() != null) {
                        agentid = msg.getAgentid();
                    } else {
                        agentid = ascsDao.getAgentidFromThirdPartyid(msg.getThirdpartyid());
                        msg.setAgentid(agentid);
                    }
                }


                // ** 以下为消息的特殊处理 **
                // 程序运行到此处，agentid仍为null，说明在注册设备完成之前收到了其他上报的消息，丢弃
                if (agentid == null) continue;
                String key = agenttype + ":" + infotype + ":" + agentid;
                // 收到方案消息，发布redis订阅，并根据用户权限，发布ws消息
                if (infotype.equals("status/pattern")) {
                    // 计算阶段倒计时
                    calcStageInfo(msg);
                    JsonElement data = msg.getData();
                    StatusPattern statusPattern = gson.fromJson(data, StatusPattern.class);
                    int control = statusPattern.getControl();
                    // 是否恢复上一条故障记录
                    if (needRecoveryFault(agentid, control)) {
                        // 故障恢复
                        Integer savedControl = lightStatusFaultAgent.get(agentid);
                        updateFaultRenewTime(agentid, (int)convertControlToFaultType(savedControl));
                        // 移除
                        lightStatusFaultAgent.remove(agentid);
                    }
                    // 发送事件通知
                    if (needSendSystemEvent(msg)) {
                        String deviceName = null;
                        AscsBaseModel ascsFromCache = ascsDao.getAscsFromCache(agentid);
                        if (ascsFromCache != null) {
                            deviceName = ascsFromCache.getName();
                        } else {
                            AscsBaseModel ascsByID = ascsDao.getAscsByID(agentid);
                            deviceName = ascsByID.getName();
                        }
                        int eventDetailsType = getDetailsTypeByControl(control);
                        sendSystemEvent(agentid,deviceName,eventDetailsType);
                        // 特殊灯色记录故障
                        recordSpecialLightStatusFault(agentid, control);
                        // 记录发生故障的路口
                        lightStatusFaultAgent.put(agentid, control);
                    }
                    // 更新缓存
                    saveAndPublishMsg(msg,key,0);
                    ascsDao.putMsgToStatusPatternMap(agentid,msg);
                    String value = gson.toJson(msg);
                    // 发布websocket订阅
                    for (Session session : patternWebSocketSet.keySet()) {
                        WebSocketServer webSocketServer = patternWebSocketSet.get(session);
                        Set<String> agentIds = webSocketServer.patternAgentidSet;
                        if (webSocketServer.subscribeAllDev()) {
                            webSocketServer.sendMessage(value);
                        } else if (agentIds != null && agentIds.contains(agentid)) {
                            webSocketServer.sendMessage(value);
                        }
                    }
                    // 消息推送到kafka
                    pushMessageToKafka(statusPatternPushTopic,msg);
                    // 保存实时方案状态
                    String savePatternSql = getSavePatternSql(msg);
                    if (StringUtils.hasLength(savePatternSql)) {
                        patternsqlqueue.offer(savePatternSql);
                    }
                }
                // 上报的检测器状态
                else if ( infotype.equals(detectorstatus) ){
                    try {
                        msg = devService.processDetectorStatus(msg);
                    } catch (Exception e) {
                        logger.warning("preProcess " + msg.getAgentid() + "'s detector status error! " + e.getMessage());
                        continue;
                    }
                    String value = gson.toJson(msg);
                    stringRedisTemplate.opsForValue().set(key, value);
                }
                else if(infotype.equals("status/channel")){ // 通道状态消息，不处理
                    saveAndPublishMsg(msg,key,0);
                }

                // 收到故障消息，发布故障消息，并保存到数据库中
                else if (infotype.equals("status/fault")) {

                    // 写入数据前，判断故障是否已处理或者已忽略，若是则不写入Redis,直接从MSg对象中删除这部分数据
                    msg = faultService.filterMsg(msg);
                    Fault[] m_faultDeque = gson.fromJson(msg.getData().getAsJsonObject().getAsJsonArray("m_FaultDeque"), Fault[].class);
                    if (m_faultDeque.length == 0)
                        continue;
                    else
                        faultService.processFaultMessage(msg);
                    // 增加name字段
                    JsonObject finalMsg = gson.toJsonTree(msg).getAsJsonObject();
                    AscsBaseModel ascsFromCache = ascsDao.getAscsFromCache(agentid);
                    if (ascsFromCache != null) {
                        String name = ascsFromCache.getName();
                        finalMsg.addProperty("name", name);
                    }
                    saveAndPublishMsg(msg,key, 0);

                }
                // 收到流量消息，保存到数据库中
                else if (infotype.equals("status/currentvolume")) {
                    historyDataDao.SaveFlowData(msg);
                    // 计算路口和相位统计数据，保存到redis以及数据库
                    processFlowData(msg);
                }
                // 收到车道流量消息
                else if (infotype.equals("status/crosstrafficdata")) {
                    // 计算获取全息车道ID
                    processCrossTrafficData(msg);
                    // 消息推送到kafka
                    pushMessageToKafka(laneFlowPushTopic,msg);
//                    String value = gson.toJson(msg);
//                    if (isPushKafkaEnable) {
//                        kafkaTemplate.send(laneFlowPushTopic,msg.getAgentid(),value);
//                    }
                    saveAndPublishMsg(msg,key,0);

                }
            } catch (Exception e) {
                logger.warning("taskExecutor Exception: " +  e.getMessage());
            }
        }

    }

    /**
     * 推送openatc信号机消息至kafka
     * @param topic
     * @param msg
     */
    private void pushMessageToKafka(String topic, MessageData msg) {
        if (isPushKafkaEnable) {
            String value = gson.toJson(msg);
            try {
                kafkaTemplate.send(topic,msg.getAgentid(),value);
            } catch (KafkaException e) {
                logger.warning("Enable push msg to kafka,but kafka config error!");
            }
        }
    }

    /**
     * 是否需要恢复故障
     * @param agentid
     * @param control
     * @return
     */
    private boolean needRecoveryFault(String agentid, int control) {
        Integer savedControl = lightStatusFaultAgent.get(agentid);
        if (savedControl == null) return false;
        return control != savedControl;
    }

    /**
     * 更新故障记录中设备的恢复时间
     * @param agentId
     * @param faultType
     */
    private void updateFaultRenewTime(String agentId, int faultType) {
        String selectLightFault = String.format("select agentid,m_w_faultid from fault " +
                " where agentid = '%s' and m_w_fault_type = %s and m_un_fault_renew_time = 0",agentId,faultType);
        List<Map<String, Object>> unProceedFault = jdbcTemplate.queryForList(selectLightFault);
        if (CollectionUtils.isEmpty(unProceedFault)) {
            logger.warning("Need renew fault,but can't find fault!Sql is: " + selectLightFault);
            return;
        }
//        if (!CollectionUtils.isEmpty(unProceedFault)) {
        Set<String> faultIds = new HashSet<>();
        Set<String> needUpdateDev = new HashSet<>();
        for (Map<String, Object> fault : unProceedFault) {
            String agentid = (String) fault.get("agentid");
            Long faultId = (Long) fault.get("m_w_faultid");
            needUpdateDev.add(agentid);
            faultIds.add(String.valueOf(faultId));
        }
        String updateFaultRenewTime = String.format("update fault set m_un_fault_renew_time=%s,enumerate='2',operator='admin' " +
                        " where m_w_faultid in %s and agentid in %s",
                System.currentTimeMillis() / 1000,AscsDao.getInCond(new ArrayList<>(faultIds)), AscsDao.getInCond(new ArrayList<>(needUpdateDev)));
        jdbcTemplate.update(updateFaultRenewTime);
//        }
    }
    /**
     * 黄闪、全红、关灯生成故障记录
     * @param agentId
     * @param control
     */
    private void recordSpecialLightStatusFault(String agentId, int control) {
        // 增加一条故障记录，记录设备出现特殊灯色
        Fault fault = new Fault();
        fault.setAgentid(agentId);
        Long faultId = generateFaultId(agentId);
        if (faultId == null) {
            logger.warning("Dev offline,but create faultId error! agentid: " + agentId);
            return;
        }
        fault.setM_wFaultID(faultId);
        fault.setM_byFaultBoardType(9);  // 其他故障
        fault.setMUnFaultOccurTime(System.currentTimeMillis() / 1000);
        long faultType = convertControlToFaultType(control);
        fault.setM_wFaultType(faultType);
        fault.setM_byFaultLevel(1);
        fault.setEnumerate("0");
        fault.setM_unFaultRenewTime(0l);

        faultDao.save(fault);
    }

    /**
     * 不同特殊控制对应不同故障类型
     * @param control
     * @return
     */
    private long convertControlToFaultType(int control) {
        switch (control) {
            case 1:
                return 902l;
            case 2:
                return 903l;
            case 3:
                return 904l;
        }
        throw new RuntimeException("UnSupport control convert.");
    }

    /**
     * 防止agentid_faultid不唯一，每次生成fault之前查询，如果存在则重新生成
     * 最多重复100次，如果重复100次还没有生成成功，则返回null
     * @param agentid
     * @return
     */
    public long generateFaultId(String agentid) {
        int count = 0;
        Long faultId = null;
        while (count < 100) {
            faultId = (long) new Random().nextInt(10000);
            String sql = String.format("select count(*) from fault where agentid='%s' and m_w_faultid=%s", agentid, faultId);
            int num = jdbcTemplate.queryForObject(sql, Integer.class);
            if (num == 0) {
                break;
            }
            count++;
        }
        return faultId;
    }

    /**
     * 根据控制方式决定通知的详细内容类型
     * @param control
     * @return
     */
    private int getDetailsTypeByControl(int control) {
        switch (control) {
            case 1:
                return MSG_TYPE_YELLOW_FLASH;
            case 2:
                return MSG_TYPE_ALL_RED;
            case 3:
                return MSG_TYPE_LIGHT_OFF;
        }
        throw new RuntimeException("UnSupport control!");
    }

    /**
     * 发送黄闪通知
     */
    private void sendSystemEvent(String agentId, String deviceName, int eventDetailsType) {

        SystemEvent systemEvent = new SystemEvent();

        String source = "来自系统的通知";
        String content = "";
        systemEvent.setType(TYPE_SYSTEM);
        systemEvent.setContent(content);
        systemEvent.setSource(source);
        systemEvent.setLevel(EVENT_LEVEL_URGENT);
        systemEvent.setCreatetime(DateUtil.date2str(new Date()));
        SystemEventDetails details = new SystemEventDetails();
        details.setAgentid(agentId);
        details.setMsgtype(eventDetailsType);
        details.setName(deviceName);
        systemEvent.setDetails(details);
        webSocketServer.sendSystemEvent(systemEvent);
    }

    /**
     * 是否为黄闪
     * @param msg
     * @return
     */
    private boolean needSendSystemEvent(MessageData msg) {
        JsonElement data = msg.getData();
        StatusPattern statusPattern = gson.fromJson(data, StatusPattern.class);
        int control = statusPattern.getControl();
        // 如果保存了故障，说明上一个故障还没清除，不上报
        Integer savedControl = lightStatusFaultAgent.get(msg.getAgentid());
        if (savedControl == null) {
            return control == 1 || control == 2 || control == 3;
        }
        return false;
    }

    // 车道流量处理
    private void processCrossTrafficData(MessageData msg){
        List<Lane> laneList =  featureService.getLane(msg.getAgentid());

        // 创建 车道ID和全息ID的对应关系
        Map<String,String> idMap = new HashMap<>();
        for(Lane lane : laneList){
            idMap.put( lane.getLaneno(), lane.getLaneid());
        }

        JsonObject mdo = msg.getData().getAsJsonObject();
        JsonArray datelist = mdo.get("datalist").getAsJsonArray();

        for (JsonElement laneje : datelist) {
            JsonObject lane = laneje.getAsJsonObject();

            String laneid = idMap.get( lane.get("laneno").getAsString() );
            lane.addProperty("laneid", laneid);

        }
    }

    // 计算阶段倒计时
    private void calcStageInfo(MessageData msg) {
        CountDownPattern cdPattern = AscsDao.cdPatternMap.get( msg.getAgentid() );
        if(cdPattern == null)
            return;
        JsonObject mdo = msg.getData().getAsJsonObject();
        if(mdo.get("current_stage") == null)
            return;

//        int patternid = mdo.get("patternid").getAsInt();
//        int curTime = mdo.get("curTime").getAsInt();
//        int curTime = cdPattern.getCurTime();

        // 阶段时长
        List<Integer> stages_len = new ArrayList<>();
        List<Integer> stages_seq = new ArrayList<>();

        // 获取阶段时长
        for(CountDownStage stage : cdPattern.getStages().values()){
            stages_len.add( stage.getSplit() );
            stages_seq.add( stage.getStageNum());
        }


        int current_stage = mdo.get("current_stage").getAsInt();
        int cal_current_stage = cdPattern.getCurStageNum(); // 线程计算中的当前阶段
        int current_stagecd = cdPattern.getCurStageCd(); // 线程计算的当前阶段倒计时
        // 线程计算中的当前阶段和上报的当前阶段不一致，表示当前计算的阶段已运行结束
        if( current_stage != cal_current_stage ){
            CountDownStage cdstage = cdPattern.getStages().get( current_stage );
            if( cdstage != null )
                current_stagecd = cdstage.getSplit();
        }

        // 阶段时长等于零，表示当前计算的阶段已运行结束
        if(current_stagecd == 0){
            CountDownStage cdstage = cdPattern.getStages().get( current_stage );
            if( cdstage != null )
                current_stagecd = cdstage.getSplit();
        }
        mdo.addProperty("current_stagecd",current_stagecd);
        mdo.add("stages_len",GsonUtil.getInstance().toJsonTree(stages_len));
        mdo.add("stages_seq",GsonUtil.getInstance().toJsonTree(stages_seq));
//        mdo.addProperty("stages_cur",curTime);

        logger.config( "Push status/pattern! Agentid:" + msg.getAgentid() + " curStage: " + current_stage + " stagecd: " + current_stagecd + " stages_len" + stages_len + " stages_seq" + stages_seq + " " + cdPattern );

    }

    private StatusPatternPhase getPhaseStatusFromMsg(int phaseid, List<StatusPatternPhase> statusPatternPhaseList) {

        for(StatusPatternPhase statusPatternPhase : statusPatternPhaseList){
            if( statusPatternPhase.getId() == phaseid )
                return statusPatternPhase;
        }

        return null;

    }


    /**
     * 将消息保存到redis，并发布
     * @param msg
     * @param key
     * @param expire
     */
    private void saveAndPublishMsg(MessageData msg, String key, int expire) {
        if (msg == null) return;
        String value = gson.toJson(msg);
        saveAndPublishMsg(value, key, expire);
    }
    /**
     * 将消息保存到redis，并发布
     * @param msg
     * @param key
     * @param expire
     */
    private void saveAndPublishMsg(String msg, String key, int expire) {
        if(isRedisEnable) {
            if(expire == 0)
                stringRedisTemplate.opsForValue().set(key, msg);
            else
                stringRedisTemplate.opsForValue().set(key, msg, expire, TimeUnit.SECONDS);

            stringRedisTemplate.convertAndSend(key, msg);
        }
    }


    /**
     *  @param msg
     *
     */
    private void processFlowData(MessageData msg) {
        String agentid = msg.getAgentid();

        JsonElement data = msg.getData();
        Volume volume = null;
        try {
            volume = gson.fromJson(data, Volume.class);
        } catch (JsonSyntaxException e){
            logger.warning("Json parse Exception: " + e.getMessage());
        }

        // 数据为空，不进行计算
        if(volume == null || volume.getDetector() == null || volume.getDetector().isEmpty())
            return;

        // 获取相位参数、检测器参数
        RESTRet<JsonObject> restRet;
        try {
            restRet = (RESTRet<JsonObject>)templateController.getTemplate(agentid);
        } catch (Exception e) {
            logger.warning(agentid + " request feature Exception: " + e.getMessage() + "，Update statistics data failed!");
            return;
        }

        if (restRet.isSuccess()) {
            try {
                JsonObject feature = restRet.getData();
                devService.savePhasesAndInteStatistics(feature,agentid, volume);
            } catch (Exception e){
                e.printStackTrace();
                logger.warning(agentid + " save statistics data Exception: " + e.getMessage());
            }
        } else {
            logger.warning("Request feature error: " + restRet.toString());
        }
    }

    /**
     * 拼接保存方案的sql
     * @param msg
     * @return
     */
    private String getSavePatternSql(MessageData msg) {
        JsonObject msgdata = new JsonObject();
//        JsonArray msgphase = new JsonArray();

        try {
            // 删除不需要保存的数据
            JsonObject mdo = msg.getData().getAsJsonObject();

            msgdata.add("mode",mdo.get("mode"));
            msgdata.add("control",mdo.get("control"));
            msgdata.add("patternid",mdo.get("patternid"));
            msgdata.add("cycle",mdo.get("cycle"));
            msgdata.add("patternoffset",mdo.get("patternoffset"));
            msgdata.add("current_stage",mdo.get("current_stage"));
            msgdata.add("current_phase",mdo.get("current_phase"));


//            JsonArray phases = mdo.get("phase").getAsJsonArray();
//            for (JsonElement phase : phases) {
//                JsonObject phosedatanew = new JsonObject();
//
//                JsonObject phosedata = phase.getAsJsonObject();
//                phosedatanew.add("id",phosedata.get("id"));
//                phosedatanew.add("type",phosedata.get("type"));
//                phosedatanew.add("pedtype",phosedata.get("pedtype"));
//
//                msgphase.add(phosedatanew);
//            }
        } catch (Exception e){
            logger.warning("Deal Pattern Data error!:" + e.getMessage());
            return null;
        }

        String sql = String.format("INSERT INTO pattern_realtime (agentid,time,data) VALUES " +
                        "('%s','%s','%s')",
                msg.getAgentid(),msg.getCreatetime(),msgdata.toString());
        return sql;
    }

}
